package com.pkk.spring.cloud.web.runner.controller.kafka.consumer;

import com.alibaba.fastjson.JSONObject;
import com.pkk.spring.cloud.web.runner.controller.kafka.constant.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import static com.pkk.spring.cloud.web.runner.controller.kafka.constant.KafkaConstant.OTHER_SIMPLE_TOPIC;

/**
 * @description: kafka的简单接收器
 * @author: peikunkun
 * @create: 2021-05-24 17:15
 **/
@Slf4j
@Component
public class KafkaSimpleConsumer {


    /**
     * 消息消费
     *
     * @param message
     * @author peikunkun
     * @date 2021/5/24 17:20
     */
    @KafkaListener(topics = {KafkaConstant.SIMPLE_TOPIC})
    public void simpleConsumer(ConsumerRecord<?, ?> message) {
        log.info("【simpleConsumer】-接收到的消息为：topic:[{}]-partition:[{}]-value:[{}]-ALL:【{}】", message.topic(), message.partition(), message.value(), JSONObject.toJSONString(message));
    }


    /**
     * 消息消费
     *
     * @param message
     * @author peikunkun
     * @date 2021/5/24 17:20
     */
    @KafkaListener(topics = {KafkaConstant.SIMPLE_TOPIC})
    public void simpleConsumer1(ConsumerRecord<?, ?> message) {
        log.info("【simpleConsumer1】-接收到的消息为：topic:[{}]-partition:[{}]-value:[{}]-ALL:【{}】", message.topic(), message.partition(), message.value(), JSONObject.toJSONString(message));
    }

    /**
     * 消息消费
     *
     * @param message
     * @author peikunkun
     * @date 2021/5/24 17:20
     */
    @KafkaListener(topics = {KafkaConstant.SIMPLE_TOPIC}, groupId = "otherKafkaGroupId")
    public void simpleConsumer2(ConsumerRecord<?, ?> message) {
        log.info("【simpleConsumer2】-接收到的消息为：topic:[{}]-partition:[{}]-value:[{}]-ALL:【{}】", message.topic(), message.partition(), message.value(), JSONObject.toJSONString(message));
    }

    /**
     * 消息消费
     *
     * @param message
     * @author peikunkun
     * @date 2021/5/24 17:20
     */
    @KafkaListener(topics = {OTHER_SIMPLE_TOPIC})
    public void simpleConsumer3(ConsumerRecord<?, ?> message) {
        log.info("【simpleConsumer3】-接收到的消息为：topic:[{}]-partition:[{}]-value:[{}]-ALL:【{}】", message.topic(), message.partition(), message.value(), JSONObject.toJSONString(message));
    }
}
